package com.deepglint.source;

import com.deepglint.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

public class SourceTest_Collection {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置全局并行度
        env.setParallelism(1);

        // 从集合中读取数据
        DataStream<SensorReading> dataSource = env.fromCollection(Arrays.asList(
                new SensorReading("sensor1", "name1", 123456L, 34.5),
                new SensorReading("sensor2", "name2", 123457L, 35.5),
                new SensorReading("sensor3", "name3", 123458L, 36.5),
                new SensorReading("sensor4", "name4", 123459L, 37.5)
        ));

        // 传入没有元素
        DataStreamSource<Integer> inputDateSource = env.fromElements(1, 2, 3, 5, 6);

        // 打印输出
        // 可以指定流的名称
        dataSource.print("data");
        inputDateSource.print("int");

        // 执行
        // 可以指定那个的流任务执行，通过传递jobName参数
        // 如果没有传递，则执行全部
        env.execute();
    }
}
